package com.lock.util;

import java.net.URL;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.apache.commons.configuration.HierarchicalConfiguration.Node;
import org.apache.commons.configuration.ConfigurationUtils;
import org.apache.commons.configuration.FileSystem;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.commons.configuration.tree.ConfigurationNode;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.google.common.net.HostAndPort;

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.sync.RedisClusterCommands;

public class LockUtilsConfig {

    private final static Logger log = LoggerFactory.getLogger(LockUtilsConfig.class);

    private static final String VERSION_NAM = "version";
    private static final String VERSION_VAL_2_0_0 = "2.0.0";
    public static final String UNDER_LINE = "_";

    public static boolean isInited = false;

    public static boolean CONFIG_OK = false;
    public static String CONFIG_VERSION = null;
    public static String GOLBAL_LOCK_NAME_PREFIX = "";

    public static boolean PERMIT_RLOCK = false;
    public static boolean RLOCK_IS_CLUSTER = false;
    public static int RLOCK_AUTO_TIMEOUT_MS = -1;
    public static int RLOCK_CMD_TIME_OUT_MS = 1000;
    public static List<HostAndPort> RLOCK_SVR_LIST = Lists.newArrayList();
    private static StatefulRedisClusterConnection<String, String> clusterRedisConn = null;
    private static StatefulRedisConnection<String, String> singleRedisConn = null;

    public static boolean PERMIT_ZLOCK = false;
    public static int ZLOCK_SESSION_TIMEOUT_MS = 60000;
    public static int ZLOCK_CONNECTION_TIMEOUT_MS = 15000;
    public static int ZLOCK_MAX_RETRY_ATTEMPTS = 3;
    public static int ZLOCK_RETRY_SLEEP_MS = 1000;
    // public static int ZLOCK_AUTO_TIMEOUT_MS = 60 * 60 * 1000;
    public static String ZLOCK_BASE_PATH = "daling_zk";
    public static String ZLOCK_PREFIX_MID_PATH = "zklocking2";
    public static String ZLOCK_CONN_STR = null;
    public static List<HostAndPort> ZLOCK_SVR_LIST = Lists.newArrayList();
    private static CuratorFramework zkclient = null;

    static {
        initConfig();
        isInited = true;
    }

    /**
     * @return CuratorFramework
     */
    public static CuratorFramework getZKClient() {
        return zkclient;
    }

    public static RedisClusterCommands<String, String> getRedisCmd() {
        if (RLOCK_IS_CLUSTER) {
            if (clusterRedisConn.isOpen()) {
                return clusterRedisConn.sync();
            }
        } else {
            if (singleRedisConn.isOpen()) {
                return singleRedisConn.sync();
            }
        }
        return null;
    }

    /**
     *
     */
    public static void releaseZKClient() {
        log.warn("Close Curator ZKCient");
        try {
            if (zkclient != null) {
                CloseableUtils.closeQuietly(zkclient);
            }
        } catch (Throwable e) {
            // ignore
        }
        zkclient = null;
    }

    public static void releaseRedisClient() {
        log.warn("Close Lettuce RedisCient");
        try {
            if (clusterRedisConn != null && clusterRedisConn.isOpen()) {
                clusterRedisConn.close();
            }
            clusterRedisConn = null;

        } catch (Throwable e) {
            // ignore
        }
        try {
            if (singleRedisConn != null && singleRedisConn.isOpen()) {
                singleRedisConn.close();
            }
            singleRedisConn = null;
        } catch (Throwable e) {
            // ignore
        }
    }

    private static synchronized void initConfig() {
        FileSystem fileSystem = FileSystem.getDefaultFileSystem();
        URL tmpURL = null;

        tmpURL = ConfigurationUtils.locate(fileSystem, null, "lock2s.xml");
        if (tmpURL != null) {
            System.err.printf("已找到配置文件:lock2s.xml, 文件位置:%s\n", tmpURL.getFile());
            System.err.printf("开始初始化LockUtils配置...\n");
            initLock2SXmlConfig();
        } else {
            System.err.println("未找到配置文件:lock2s.xml, 继续查找zk2s.properties(ZK配置), redis2s.xml");
            {// ZK
                tmpURL = ConfigurationUtils.locate(fileSystem, null, "zk2s.properties");
                if (tmpURL != null) {
                    System.err.printf("已找到配置文件:zk2s.properties, 文件位置:%s\n", tmpURL.getFile());
                    System.err.printf("开始初始化LockUtils ZK部分配置...\n");
                    initZK2SPropsConfig();
                } else {
                    System.err.println("未找到配置文件:zk2s.properties, 缺失ZK锁功能");
                }
            }
            {// Redis
                tmpURL = ConfigurationUtils.locate(fileSystem, null, "redis2s.xml");
                if (tmpURL != null) {
                    System.err.printf("已找到配置文件:redis2s.xml, 文件位置:%s\n", tmpURL.getFile());
                    System.err.printf("开始初始化LockUtils Redis部分配置...\n");
                    initRedis2SXmlConfig();
                } else {
                    System.err.println("未找到配置文件:redis2s.xml或redis2s.properties, 缺失Redis锁功能");
                }
            }
        }
        // 所有属性完成配置文件读取后, 统一生成ZK和Redis资源
        initClientResource();
        // 如果判断RLock和ZLock配置均缺失, 则抛出异常
        StringBuilder sb = new StringBuilder();
        sb.append('\n');
        sb.append("###########################################################\n");
        sb.append("LockUtils\n");
        sb.append("  ├ CONFIG_OK? :%b\n");
        sb.append("  ├ CONFIG_VERSION :%s\n");
        sb.append("  ├ GOLBAL_LOCK_NAME_PREFIX :%s\n");
        sb.append("  ├ PERMIT_RLOCK :%b\n");
        sb.append("  │  ├ RLOCK_IS_CLUSTER :%s\n");
        sb.append("  │  ├ RLOCK_AUTO_TIMEOUT_MS :%s\n");
        sb.append("  │  ├ RLOCK_CMD_TIME_OUT_MS :%s\n");
        sb.append("  │  └ RLOCK_SERVERS :%s\n");
        sb.append("  └ PERMIT_ZLOCK :%b\n");
        // sb.append(" ├ ZLOCK_AUTO_TIMEOUT_MS :%s\n");
        sb.append("     ├ ZLOCK_SESSION_TIMEOUT_MS :%s\n");
        sb.append("     ├ ZLOCK_MAX_RETRY_ATTEMPTS :%s\n");
        sb.append("     ├ ZLOCK_RETRY_SLEEP_MS :%s\n");
        sb.append("     ├ ZLOCK_BASE_PATH :%s\n");
        sb.append("     ├ ZLOCK_PREFIX_MID_PATH :%s\n");
        sb.append("     ├ ZLOCK_CONNECTION_TIMEOUT_MS :%s\n");
        sb.append("     ├ ZLOCK_CONN_STR :%s\n");
        sb.append("     └ ZLOCK_SERVERS:%s\n");
        sb.append("###########################################################\n\n");
        System.err.printf(sb.toString(),
                CONFIG_OK,
                CONFIG_OK ? CONFIG_VERSION : "--",
                CONFIG_OK ? GOLBAL_LOCK_NAME_PREFIX : "--",

                PERMIT_RLOCK,
                PERMIT_RLOCK ? RLOCK_IS_CLUSTER : "--",
                PERMIT_RLOCK ? RLOCK_AUTO_TIMEOUT_MS : "--",
                PERMIT_RLOCK ? RLOCK_CMD_TIME_OUT_MS : "--",
                PERMIT_RLOCK ? RLOCK_SVR_LIST : "--",

                PERMIT_ZLOCK,
                // PERMIT_ZLOCK ? ZLOCK_AUTO_TIMEOUT_MS : "--",
                PERMIT_ZLOCK ? ZLOCK_SESSION_TIMEOUT_MS : "--",
                PERMIT_ZLOCK ? ZLOCK_MAX_RETRY_ATTEMPTS : "--",
                PERMIT_ZLOCK ? ZLOCK_RETRY_SLEEP_MS : "--",
                PERMIT_ZLOCK ? ZLOCK_BASE_PATH : "--",
                PERMIT_ZLOCK ? ZLOCK_PREFIX_MID_PATH : "--",
                PERMIT_ZLOCK ? ZLOCK_CONNECTION_TIMEOUT_MS : "--",
                PERMIT_ZLOCK ? ZLOCK_CONN_STR : "--",
                PERMIT_ZLOCK ? ZLOCK_SVR_LIST : "--"

        );

    }

    private static void initClientResource() {
        if (!CONFIG_OK || (!PERMIT_RLOCK && !PERMIT_ZLOCK)) {
            throw new RuntimeException("CONFIG NOT OK");
        }
        if (PERMIT_RLOCK) {
            if (RLOCK_IS_CLUSTER) {
                Set<RedisURI> redisServerSet = new HashSet<>();
                for (HostAndPort hostPort : RLOCK_SVR_LIST) {
                    redisServerSet.add(new RedisURI(hostPort.getHostText(), hostPort.getPort(), Duration.ofMillis(RLOCK_CMD_TIME_OUT_MS)));
                }
                RedisClusterClient clusterClient = RedisClusterClient.create(redisServerSet);
                clusterRedisConn = clusterClient.connect();
                singleRedisConn = null;
            } else {
                HostAndPort hostPort = RLOCK_SVR_LIST.get(0);
                RedisURI singleRedisURI = new RedisURI(hostPort.getHostText(), hostPort.getPort(), Duration.ofMillis(RLOCK_CMD_TIME_OUT_MS));
                RedisClient lettuceRedisClient = RedisClient.create(singleRedisURI);
                singleRedisConn = lettuceRedisClient.connect();
                clusterRedisConn = null;
            }
        }
        if (PERMIT_ZLOCK) {
            zkclient = CuratorFrameworkFactory.newClient(ZLOCK_CONN_STR, ZLOCK_SESSION_TIMEOUT_MS, ZLOCK_CONNECTION_TIMEOUT_MS,
                    new ExponentialBackoffRetry(0, ZLOCK_MAX_RETRY_ATTEMPTS, ZLOCK_RETRY_SLEEP_MS));
            zkclient.start();

        }

    }

    /**
     *
     */
    private static void initRedis2SXmlConfig() {
        String DEFAULT_POOL_NAME = null;
        String version = "";
        try {
            XMLConfiguration config = new XMLConfiguration();
            config.setEncoding("UTF-8");
            config.setDelimiterParsingDisabled(false);
            config.setThrowExceptionOnMissing(true);
            config.load("redis2s.xml");

            Node node = config.getRoot();
            for (ConfigurationNode confNode : node.getAttributes()) {
                log.trace("confNode.getName():{}", confNode.getName());
                log.trace("confNode.getValue():{}", confNode.getValue());
                if (StringUtils.equals(confNode.getName(), VERSION_NAM)) {
                    version = (String) confNode.getValue();
                }
            }
            log.trace("root:{}, version:{}", node.getName(), version);

            String deffaultPoolName = config.getString("default-connection-pool");
            if (StringUtils.isNotBlank(deffaultPoolName)) {
                log.trace("DEFAULT_POOL_NAME={}", deffaultPoolName);
                DEFAULT_POOL_NAME = deffaultPoolName;
            }
            int connPoolQty = config.getMaxIndex("connection-pool") + 1;
            log.trace("ConnPoolQty:{}", connPoolQty);
            if (connPoolQty <= 0) {
                DEFAULT_POOL_NAME = null;
                throw new RuntimeException("connection-pool is null");
            }

            for (int i = 0; i < connPoolQty; i++) {
                String poolName = config.getString("connection-pool(" + i + ")[@name]", "");
                if (StringUtils.isBlank(poolName)) {
                    throw new IllegalArgumentException("poolName can not be empty");
                }
                if (i == 0 && StringUtils.isBlank(DEFAULT_POOL_NAME)) {
                    log.trace("DEFAULT_POOL_NAME={}", poolName);
                    DEFAULT_POOL_NAME = poolName;
                }
                //
                if (!StringUtils.equals(DEFAULT_POOL_NAME, poolName)) {
                    continue;// 只使用默认的Redis连接池配置
                } else {
                    PERMIT_RLOCK = true;
                }

                String poolType = config.getString("connection-pool(" + i + ")[@type]", "");
                if (StringUtils.isBlank(poolType)) {// 2.0.0版本以下不配置, 继续查找connection-pool.pool-type
                    poolType = config.getString("connection-pool(" + i + ").pool-type", "");
                }
                if (StringUtils.equals(poolType, "SINGLE")) {
                    RLOCK_IS_CLUSTER = false;
                } else if (StringUtils.equals(poolType, "CLUSTER")) {
                    RLOCK_IS_CLUSTER = true;
                }
                RLOCK_AUTO_TIMEOUT_MS = 6000;

                if (!RLOCK_IS_CLUSTER) {
                    RLOCK_CMD_TIME_OUT_MS = config.getInt("connection-pool(" + i + ").timeout", 1000);
                    String host = config.getString("connection-pool(" + i + ").server.host");
                    int port = config.getInt("connection-pool(" + i + ").server.port");
                    RLOCK_SVR_LIST.add(HostAndPort.fromParts(host, port));
                } else {
                    RLOCK_CMD_TIME_OUT_MS = config.getInt("connection-pool(" + i + ").timeout", 1000);
                    int serversQty = config.getMaxIndex("connection-pool(" + i + ").servers.server") + 1;
                    for (int m = 0; m < serversQty; m++) {
                        String host = config.getString("connection-pool(" + i + ").servers.server(" + m + ").host");
                        int port = config.getInt("connection-pool(" + i + ").servers.server(" + m + ").port");
                        RLOCK_SVR_LIST.add(HostAndPort.fromParts(host, port));
                    }
                }
            }
            CONFIG_OK = true;
        } catch (Throwable exc) {
            PERMIT_RLOCK = false;
            RuntimeException e = new RuntimeException("Init Name...ERROR", exc);
            log.error("initRedis2SXmlConfig error:{}", e);
            throw e;
        } finally {
            log.info("DEFAULT_POOL_NAME={} ", DEFAULT_POOL_NAME);
        }
    }

    private static void initZK2SPropsConfig() {
        try {
            PropertiesConfiguration config = new PropertiesConfiguration();
            config.setDelimiterParsingDisabled(true);
            config.setEncoding("UTF-8");
            config.setThrowExceptionOnMissing(true);
            config.load("zk2s.properties");
            List<String> toList = Lists.newArrayList();
            {
                String TMP_ZK_CONN_STR = config.getString("zk.server.list");
                System.err.printf("配置的ZK_CONN_STR:%s\n", TMP_ZK_CONN_STR);
                String[] tmpZkConnStrArray = StringUtils.split(TMP_ZK_CONN_STR, ',');

                List<String> fromList = Lists.newArrayList(tmpZkConnStrArray);
                Random rd = new Random();
                for (int i = tmpZkConnStrArray.length; i > 0; i--) {
                    String zkConnStr = fromList.remove(rd.nextInt(i));
                    ZLOCK_SVR_LIST.add(HostAndPort.fromString(zkConnStr));
                    toList.add(zkConnStr);
                }
            }
            ZLOCK_CONN_STR = StringUtils.join(toList.toArray(), ',');
            ZLOCK_BASE_PATH = config.getString("base.path", "/daling_zk");
            ZLOCK_SESSION_TIMEOUT_MS = config.getInt("session_timeout_ms", 60 * 1000);
            ZLOCK_CONNECTION_TIMEOUT_MS = config.getInt("conn_timeout_ms", 15 * 1000);
            ZLOCK_MAX_RETRY_ATTEMPTS = config.getInt("retry_max_qty", 3);
            ZLOCK_RETRY_SLEEP_MS = config.getInt("retry_sleep_ms", 1000);
            PERMIT_ZLOCK = true;
            CONFIG_OK = true;
        } catch (org.apache.commons.configuration.ConfigurationException exc) {
            PERMIT_ZLOCK = false;
            log.error("GlobalConfigurationException", exc);
            throw new RuntimeException(exc);
        }
    }

    /**
     *
     */
    private static void initLock2SXmlConfig() {
        try {
            XMLConfiguration config = new XMLConfiguration();
            config.setEncoding("UTF-8");
            config.setDelimiterParsingDisabled(false);
            config.setThrowExceptionOnMissing(true);
            config.load("lock2s.xml");

            Node rootnode = config.getRoot();
            for (ConfigurationNode confNode : rootnode.getAttributes()) {
                log.trace("confNode.getName():{}", confNode.getName());
                log.trace("confNode.getValue():{}", confNode.getValue());
                if (StringUtils.equals(confNode.getName(), VERSION_NAM)) {
                    CONFIG_VERSION = (String) confNode.getValue();
                }
            }
            log.trace("root:{}, version:{}", rootnode.getName(), CONFIG_VERSION);

            if (!VERSION_VAL_2_0_0.equals(CONFIG_VERSION)) {
                throw new RuntimeException("root tag lock2s-configuration propertiy version must be 2.0.0");
            } else {
                initLockUtilsConfigV2_0_0(config);
            }

        } catch (Throwable exc) {
            RuntimeException e = new RuntimeException("Init ...ERROR", exc);
            throw e;
        } finally {

        }

    }

    /**
     * @param config
     * @return
     */

    private static void initLockUtilsConfigV2_0_0(XMLConfiguration config) {
        try {
            GOLBAL_LOCK_NAME_PREFIX = config.getString("lock-name-prefix", "");
            log.trace("通用锁Key前缀:{}", GOLBAL_LOCK_NAME_PREFIX);

            {// 处理Redis锁
                int rLockQty = config.getMaxIndex("rlock-config") + 1;
                log.trace("rLockQty:{}", rLockQty);
                if (rLockQty == 1) {
                    PERMIT_RLOCK = true;

                    RLOCK_CMD_TIME_OUT_MS = config.getInt("rlock-config.command-timeout", 1000);
                    log.trace("指令超时时间:{}", RLOCK_CMD_TIME_OUT_MS);

                    RLOCK_AUTO_TIMEOUT_MS = config.getInt("rlock-config.lock-auto-timeout", 60000);
                    log.trace("默认RLock锁超时时间:{}", RLOCK_AUTO_TIMEOUT_MS);

                    {
                        int redisClusterQty = config.getMaxIndex("rlock-config.servers.server") + 1;
                        log.trace("redisClusterQty={}", redisClusterQty);
                        if (redisClusterQty > 0) {// RedisCluster
                            RLOCK_IS_CLUSTER = true;
//							Set<RedisURI> redisServerSet = new HashSet<>();
                            for (int i = 0; i < redisClusterQty; i++) {
                                String host = config.getString("rlock-config.servers.server(" + i + ").host");
                                int port = config.getInt("rlock-config.servers.server(" + i + ").port");
                                RLOCK_SVR_LIST.add(HostAndPort.fromParts(host, port));
                                // redisServerSet.add(new RedisURI(host, port, Duration.ofMillis(RLOCK_CMD_TIME_OUT_MS)));
                                // log.trace("host={},port={}", host, port);
                            }
//							RedisClusterClient clusterClient = RedisClusterClient.create(redisServerSet);
//							clusterRedisConn = clusterClient.connect();
//							singleRedisConn = null;

                        } else {// 单机
                            String host = config.getString("rlock-config.server.host");
                            int port = config.getInt("rlock-config.server.port");
                            RLOCK_SVR_LIST.add(HostAndPort.fromParts(host, port));
//							RedisURI singleRedisURI = new RedisURI(host, port, Duration.ofMillis(RLOCK_CMD_TIME_OUT_MS));
//							log.trace("host={},port={}", host, port);
//							RedisClient lettuceRedisClient = RedisClient.create(singleRedisURI);
//							singleRedisConn = lettuceRedisClient.connect();
//							clusterRedisConn = null;
                        }
                    }
                    int redisServerQty = RLOCK_SVR_LIST.size();
                    log.trace("Redis服务器数量:{}, {}", redisServerQty, RLOCK_IS_CLUSTER ? "Redis集群锁" : "Redis单机锁");
                    int idx = 0;
                    for (HostAndPort hostPort : RLOCK_SVR_LIST) {
                        log.trace("Redis服务器[{}]:host={},port={},command-timeout={}", idx++, hostPort.getHostText(), hostPort.getPort(), RLOCK_CMD_TIME_OUT_MS);
                    }

                } else if (rLockQty > 1) {
                    throw new IllegalArgumentException("大于一个 RLock配置[" + rLockQty + "]");
                }
            }
            {// 处理ZK锁
                int zLockQty = config.getMaxIndex("zlock-config") + 1;
                log.trace("zLockQty:{}", zLockQty);
                if (zLockQty == 1) {
                    PERMIT_ZLOCK = true;
                    int serverClusterQty = config.getMaxIndex("zlock-config(0).servers.server") + 1;
                    if (serverClusterQty % 2 != 1) {
                        throw new IllegalArgumentException("ZK服务器数量不是奇数:" + serverClusterQty);
                    }

                    List<String> fromSvrList = Lists.newArrayList();
                    for (int i = 0; i < serverClusterQty; i++) {
                        String host = config.getString("zlock-config.servers.server(" + i + ").host");
                        int port = config.getInt("zlock-config.servers.server(" + i + ").port");
                        fromSvrList.add(host + ":" + port);
                        ZLOCK_SVR_LIST.add(HostAndPort.fromParts(host, port));
                    }

                    Random rd = new Random();
                    List<String> toSvrList = Lists.newArrayList();
                    for (int i = serverClusterQty; i > 0; i--) {
                        String zkConnStr = fromSvrList.remove(rd.nextInt(i));
                        toSvrList.add(zkConnStr);
                    }
                    ZLOCK_CONN_STR = StringUtils.join(toSvrList.toArray(), ',');

                    ZLOCK_SESSION_TIMEOUT_MS = config.getInt("zlock-config.session-timeout", 60000);
                    ZLOCK_CONNECTION_TIMEOUT_MS = config.getInt("zlock-config.conn-timeout", 15000);
                    ZLOCK_MAX_RETRY_ATTEMPTS = config.getInt("zlock-config.max-retry-attempts", 3);
                    ZLOCK_RETRY_SLEEP_MS = config.getInt("zlock-config.retry-sleep", 1000);
                    ZLOCK_BASE_PATH = config.getString("zlock-config.base_path", "daling_zk");
                    ZLOCK_PREFIX_MID_PATH = config.getString("zlock-config.prefix_mid_path", "zklocking2");
                    // ZLOCK_AUTO_TIMEOUT_MS = config.getInt("zlock-config.lock-auto-timeout", 60 * 60 * 1000);

                    // zkclient = CuratorFrameworkFactory.newClient(ZLOCK_CONN_STR, ZLOCK_SESSION_TIMEOUT_MS, ZLOCK_CONNECTION_TIMEOUT_MS,
                    // new ExponentialBackoffRetry(0, ZLOCK_MAX_RETRY_ATTEMPTS, ZLOCK_RETRY_SLEEP_MS));
                    // zkclient.start();

                } else if (zLockQty > 1) {
                    throw new IllegalArgumentException("大于一个 ZLock配置[" + zLockQty + "]");
                }
            }
            CONFIG_OK = true;
        } catch (Exception e) {
            CONFIG_OK = false;
        }
    }

    public static void main(String[] args) throws Exception {
        boolean bb = LockUtilsConfig.CONFIG_OK;
        TimeUnit.SECONDS.sleep(1);
        log.trace("CONFIG_OK:{}", bb);
        log.trace("EXIT");
    }

}
